# -*- codeing: utf-8 -*-
'''
# @Time     :2022/1/7 15:35
# @Author   :谢怡萍
#@File      :mysql_kafka.py
#@Software  : PyCharm
'''
import datetime
import json
from kafka import KafkaProducer
from pymysqlreplication import BinLogStreamReader
from pymysqlreplication.row_event import (
    DeleteRowsEvent,
    UpdateRowsEvent,
    WriteRowsEvent
)


class DateEncoder(json.JSONEncoder):
    """
    自定义类，解决报错：
    TypeError: Object of type 'datetime' is not JSON serializable
    """
    def default(self, obj):
        if isinstance(obj, datetime.datetime):
            return obj.strftime('%Y-%m-%d %H:%M:%S')

        elif isinstance(obj, datetime.date):
            return obj.strftime("%Y-%m-%d")

        else:
            return json.JSONEncoder.default(self, obj)

# 配置数据库信息
mysql_settings = {
    'host': '127.0.0.1',
    'port': 3306,
    'user': 'root',
    'passwd': '123456'
}
def main():
    conf = {
        'bootstrap_servers': ["××××××××1:9092", "××××××××2:9092", "××××××××3:9092"],
        'topic_name': 'orderdetail',
    }
    # 创建producer
    producer = KafkaProducer(bootstrap_servers=conf['bootstrap_servers'],api_version=(0, 10))
    # 实例化binlog 流对象
    stream = BinLogStreamReader(
        connection_settings=mysql_settings,
        server_id=3,  # slave标识，唯一
        blocking=True,  # 阻塞等待后续事件
        # 设定只监控写操作：增、删、改
        only_events=[
            DeleteRowsEvent,
            UpdateRowsEvent,
            WriteRowsEvent
        ]
    )
    # 向kafka传递数据，创建两个topic
    for binlogevent in stream:
        for row in binlogevent.rows:
            if binlogevent.table=="order":
                string ='|'.join(str(i)for i in list(row["values"].values()))
                producer.send('order', string.encode())
                print(string)
            if binlogevent.table=="orderdetail":
                string =','.join(str(i)for i in list(row["values"].values()))
                producer.send('orderdetail', string.encode())
                print(string)

if __name__ == '__main__':
    main()